// Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.

// 应用进程操作
package appprocess

import (
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/redisserver"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/services/event"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/beego/beego/v2/client/orm"

	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/common"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/utils/log"
)

const (
	ZombieAppProcessCountToSleep        = 50
	ZombieAppProcessCountToSleepSeconds = 1
	AppProcessBackupCountToSleep        = 50
	AppProcessBackupCountToSleepSeconds = 1
	VerifyZombieProcessSeconds          = 90
)

type AppProcessDao struct {
	sqlSession orm.Ormer
}

type AppProcessBackupDao struct {
	sqlSession orm.Ormer
}

// NewAppProcessDao new app process dao
func NewAppProcessDao(sqlsession orm.Ormer) *AppProcessDao {
	return &AppProcessDao{sqlSession: sqlsession}
}

// CreateAppProcess create app process
func (a *AppProcessDao) CreateAppProcess(ap *AppProcess) (*AppProcess, error) {
	_, err := a.sqlSession.Insert(ap)
	if err != nil {
		log.RunLogger.Errorf("[app process data service] failed to insert app process %v for %v", ap.ID, err)
		return nil, fmt.Errorf("failed to insert app process into database for %v", err)
	}

	return ap, nil
}

// UpdateAppProcess update app process
func (a *AppProcessDao) UpdateAppProcess(ap *AppProcess) (*AppProcess, error) {
	_, err := a.sqlSession.Update(ap)
	if err != nil {
		log.RunLogger.Errorf("[app process data service] failed to update app process %v for %v", ap.ID, err)
		return nil, fmt.Errorf("failed to update app process for %v", err)
	}

	return ap, nil
}

func (a *AppProcessDao) UpdateAppProcessStateAndUpdatedAt(ap *AppProcess) (*AppProcess, error) {
	_, err := a.sqlSession.Update(ap, FieldNameState, FieldNameUpdatedAt)
	if err != nil {
		log.RunLogger.Errorf("[app process data service] failed to update app process state %v for %v", ap.ID, err)
		return nil, fmt.Errorf("failed to update app process for %v", err)
	}

	return ap, nil
}

func (a *AppProcessDao) UpdateAppProcessStateAndUpdatedAtUsingOCC(ap *AppProcess, state string) error {
	sqlStr := fmt.Sprintf("update %s set STATE=?, UPDATED_AT=? where ID=? and STATE=?", TableNameAppProcess)
	result, err := a.sqlSession.Raw(sqlStr, ap.State, ap.UpdatedAt, ap.ID, state).Exec()
	if err != nil {
		return err
	}
	num, _ := result.RowsAffected()
	if num == 0 {
		log.RunLogger.Errorf("occ update state failed, because the process %s state has been changed", ap.ID)
		curState, err := a.GetAppProcessByID(ap.ID)
		if err != nil {
			return err
		}

		return fmt.Errorf("cannot transfer app process %s state from %s to %s", ap.ID, ap.State, curState.State)
	}
	log.RunLogger.Infof("transfer app process %s state from %s to %s success", ap.ID, state, ap.State)
	return nil
}

// DeleteAppProcess delete app process
func (a *AppProcessDao) DeleteAppProcess(ap *AppProcess) error {
	_, err := a.sqlSession.Delete(ap)
	if err != nil {
		log.RunLogger.Errorf("[app process data service] failed to delete app process %v for %v", ap.ID, err)
		return fmt.Errorf("failed to update app process for %v", err)
	}

	return nil
}

// GetAppProcessByID get an app process by id
func (a *AppProcessDao) GetAppProcessByID(id string) (*AppProcess, error) {
	var ap AppProcess

	cond := orm.NewCondition()
	cond = cond.And(FieldNameProcessID, id)
	err := a.sqlSession.QueryTable(&AppProcess{}).SetCond(cond).One(&ap)

	return &ap, err
}

// GetAppProcessByFleetIDAndInstanceID get app process by fleet id and instance id with sort
func (a *AppProcessDao) GetAppProcessByFleetIDAndInstanceID(fleetID, instanceID, sort string,
	offset, limit int) (*[]AppProcess, error) {
	var aps []AppProcess

	cond := orm.NewCondition()

	if fleetID != "" {
		cond = cond.And(FieldNameFleetID, fleetID)
	}
	if instanceID != "" {
		cond = cond.And(FieldNameInstanceID, instanceID)
	}

	_, err := a.sqlSession.QueryTable(&AppProcess{}).SetCond(cond).OrderBy(sort).
		Offset(offset).Limit(limit).All(&aps)

	return &aps, err
}

// GetAllActiveAppProcess get all active processes
func (a *AppProcessDao) GetAllActiveAppProcess() ([]*AppProcess, error) {
	var aps []*AppProcess
	sqlStr := fmt.Sprintf(`select * from %s where STATE="%s"`,
		TableNameAppProcess, common.AppProcessStateActive)
	_, err := a.sqlSession.Raw(sqlStr).QueryRows(&aps)
	return aps, err
}

// GetAllActiveAppProcess get all active processes
func (a *AppProcessDao) GetAllAvailableAppProcess() ([]*AppProcess, error) {
	var aps []*AppProcess
	sqlStr := fmt.Sprintf("select * from %s where STATE=? AND "+
		"SERVER_SESSION_COUNT <MAX_SERVER_SESSION_NUM", TableNameAppProcess)
	_, err := a.sqlSession.Raw(sqlStr, common.AppProcessStateActive).QueryRows(&aps)
	if err == orm.ErrNoRows {
		log.RunLogger.Infof("there is no available app process")
	}
	return aps, err
}

func (a *AppProcessDao) GetAllAvailableAppProcessByFleetID(fleetID string) ([]*AppProcess, error) {
	var aps []*AppProcess
	sqlStr := fmt.Sprintf("select * from %s where FLEET_ID=? AND STATE=? AND "+
		"SERVER_SESSION_COUNT <MAX_SERVER_SESSION_NUM", TableNameAppProcess)
	_, err := a.sqlSession.Raw(sqlStr, fleetID, common.AppProcessStateActive).QueryRows(&aps)
	if err == orm.ErrNoRows {
		log.RunLogger.Infof("there is no available app process")
	}
	return aps, err
}

// GetAvailableAppProcessByFleetID get available app processes by fleet id
func (a *AppProcessDao) GetAvailableAppProcessByFleetID(fleetID string) (AppProcess, error) {
	var ap AppProcess
	sqlStr := fmt.Sprintf("select * from %s where FLEET_ID=? AND STATE=? AND "+
		"SERVER_SESSION_COUNT <MAX_SERVER_SESSION_NUM", TableNameAppProcess)
	err := a.sqlSession.Raw(sqlStr, fleetID, common.AppProcessStateActive).QueryRow(&ap)
	if err == orm.ErrNoRows {
		log.RunLogger.Infof("there is no available app process")
	}
	return ap, err
}

// GetBusiestAndAvailableAppProcessByFleetID 用于获取可以的process
func (a *AppProcessDao) GetBusiestAndAvailableAppProcessByFleetID(fleetID string) (AppProcess, error) {
	var ap AppProcess
	sqlStr := fmt.Sprintf("select * from %s where FLEET_ID=? AND STATE=? AND SERVER_SESSION_COUNT < "+
		"MAX_SERVER_SESSION_NUM ORDER BY SERVER_SESSION_COUNT/MAX_SERVER_SESSION_NUM DESC LIMIT 1", TableNameAppProcess)

	err := a.sqlSession.Raw(sqlStr, fleetID, common.AppProcessStateActive).QueryRow(&ap)
	if err == orm.ErrNoRows {
		log.RunLogger.Infof("there is no available app process")
	}
	return ap, err
}

// GetProcessOfBusiestAndAvailableInstanceByFleetID 在有空闲server session的情况下，选择instance最繁忙的放置，
// process选择instance里面server session 最少的
func (a *AppProcessDao) GetProcessOfBusiestAndAvailableInstanceByFleetID(fleetID string) (AppProcess, error) {
	var ap AppProcess
	sqlStr := fmt.Sprintf("SELECT * from %s where FLEET_ID=? and STATE=?"+
		"INSTANCE_ID=(SELECT INSTANCE_ID From %s where FLEET_ID=? and STATE=? GROUP BY INSTANCE_ID "+
		"HAVING SUM(SERVER_SESSION_COUNT) < SUM(MAX_SERVER_SESSION_NUM) "+
		"ORDER BY SUM(SERVER_SESSION_COUNT) DESC LIMIT 1) ORDER BY SERVER_SESSION_COUNT/MAX_SERVER_SESSION_NUM "+
		"ASC limit 1", TableNameAppProcess, TableNameAppProcess)

	err := a.sqlSession.Raw(sqlStr, fleetID, common.AppProcessStateActive, fleetID,
		common.AppProcessStateActive).QueryRow(&ap)
	if err == orm.ErrNoRows {
		log.RunLogger.Infof("there is no available app process")
	}
	return ap, err
}

// GetAllFleet get all fleet
func (a *AppProcessDao) GetAllFleet() ([]string, error) {
	var fleets []string

	var rows []orm.Params
	queryStr := fmt.Sprintf("SELECT %s FROM %s GROUP BY %s", FieldNameFleetID, TableNameAppProcess, FieldNameFleetID)
	_, err := a.sqlSession.Raw(queryStr).Values(&rows)
	if err != nil {
		return nil, err
	}

	for i, _ := range rows {
		fleetID, _ := rows[i][FieldNameFleetID].(string)

		fleets = append(fleets, fleetID)
	}

	return fleets, err
}

// ProcessCounts process counts
func (a *AppProcessDao) ProcessCounts(fleetID string) (map[string]int, error) {
	processCounts := map[string]int{}

	var rows []orm.Params
	sqlStr := fmt.Sprintf(`SELECT %s, COUNT(*) as COUNT FROM %s WHERE %s="%s" GROUP BY %s`,
		FieldNameState, TableNameAppProcess, FieldNameFleetID, fleetID, FieldNameState)
	log.RunLogger.Infof(sqlStr)
	_, err := a.sqlSession.Raw(sqlStr).Values(&rows)
	if err != nil {
		return nil, err
	}

	for i, _ := range rows {
		state, _ := rows[i]["STATE"].(string)
		countStr, _ := rows[i]["COUNT"].(string)

		count, err := strconv.Atoi(countStr)
		if err != nil {
			continue
		}
		processCounts[state] = count
	}

	return processCounts, err
}

// VerifyZombieProcess 识别僵尸进程，判断条件是超过90s没有更新状态，先SELECT再根据进程ID UPDATE，避免锁整表
func (a *AppProcessDao) VerifyZombieProcess() error {
	// SELECT出三次健康检查未生效的进程，置为ERROR
	var aps []AppProcess
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE STATE=? and UPDATED_AT < DATE_SUB(NOW(), INTERVAL %d SECOND)",
		TableNameAppProcess, VerifyZombieProcessSeconds)
	zombieProcessCount, err := a.sqlSession.Raw(sqlStr, common.AppProcessStateActive).QueryRows(&aps)
	if err != nil {
		return err
	}
	// 异步修改每个进程的状态
	log.RunLogger.Infof("[verify zombie process] finish verify zombie process, would affect %v", zombieProcessCount)
	wg := sync.WaitGroup{}
	wg.Add(int(zombieProcessCount))
	for idx, ap := range aps {
		// 避免瞬间连接数过多，每更新50个进程状态停止1s
		if idx%ZombieAppProcessCountToSleep == 0 {
			time.Sleep(ZombieAppProcessCountToSleepSeconds * time.Second)
		}
		go event.RegisterProcessError("[zombie process]", event.ResourceEvent{ResourceId: ap.ID, FleetId: ap.FleetID},
			"find a zombie process", event.EventTraceStateWarning)
		go func(app_process AppProcess) {
			defer wg.Done()
			err := redisserver.DeleteProcessFromRedis(app_process.FleetID, app_process.ID)
			if err != nil {
				log.RunLogger.Errorf("[verify zombie process] delete process from redis failed after retry %v times for %v", redisserver.RedisRetryTimes, err)
			}
			sqlStrUpdateAppProcess := fmt.Sprintf("update %s SET STATE=? WHERE ID=?", TableNameAppProcess)
			rsl, err := a.sqlSession.Raw(sqlStrUpdateAppProcess, common.AppProcessStateError, app_process.ID).Exec()
			if err != nil {
				log.RunLogger.Errorf("[verify zombie process] update health check state failed for process %s, err: %+v",
					app_process.ID, err)
				return
			}
			num, err := rsl.RowsAffected()
			if err != nil {
				log.RunLogger.Errorf("[verify zombie process] get update health check state result failed for process %s, err: %+v",
					app_process.ID, err)
				return
			}
			log.RunLogger.Infof("[verify zombie process] verify zombie process %s, affect %v",
				app_process.ID, num)
		}(ap)
	}
	wg.Wait()
	return nil
}

// AppProcessBackup数据表初始化
func NewAppProcessBackupDao(sqlsession orm.Ormer) *AppProcessBackupDao {
	return &AppProcessBackupDao{sqlSession: sqlsession}
}

// InsertAppProcessBackup: 在备份表中插入数据
func (b *AppProcessBackupDao) InsertAppProcessBackup(ap *AppProcessBackup) (*AppProcessBackup, error) {
	_, err := b.sqlSession.Insert(ap)
	if err != nil {
		log.RunLogger.Errorf("[app process backup service] failed to insert app process %s to backup table error: %+v", ap.ID, err)
		return nil, err
	}
	return ap, nil
}

// DeleteAppProcessBackup: 从备份表中删除数据
func (b *AppProcessBackupDao) DeleteAppProcessBackup(ap *AppProcessBackup) error {
	_, err := b.sqlSession.Delete(ap)
	if err != nil {
		log.RunLogger.Errorf("[app process backup service] failed to delete app process %s from backup table error: %+v", ap.ID, err)
		return err
	}
	return nil
}

// QueryAppProcessBackupById: 根据id从备份表中查询数据
func (b *AppProcessBackupDao) QueryAppProcessBackupById(id string) (*AppProcessBackup, error) {
	var ap AppProcessBackup
	cond := orm.NewCondition()
	cond = cond.And(FieldNameProcessID, id)
	err := b.sqlSession.QueryTable(&AppProcessBackup{}).SetCond(cond).One(&ap)

	return &ap, err
}

// CleanAppProcessBackup: 根据更新时间从备份表中清理dayBefore天前的数据
func (b *AppProcessBackupDao) CleanAppProcessBackup(dayBefore int) error {
	var aps []AppProcessBackup
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE UPDATED_AT < DATE_SUB(NOW(), INTERVAL %d DAY)",
		TableNameAppProcessBackup, dayBefore)
	toBeCleanCount, err := b.sqlSession.Raw(sqlStr).QueryRows(&aps)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] query to be clean app process from backup table failed: %+v", err)
		return err
	}
	log.RunLogger.Infof("[express data cleaner] find %v expressed app process from backup table", toBeCleanCount)
	wg := sync.WaitGroup{}
	wg.Add(int(toBeCleanCount))
	for idx, ap := range aps {
		if idx%AppProcessBackupCountToSleep == 0 {
			time.Sleep(AppProcessBackupCountToSleepSeconds * time.Second)
		}
		go func(app_process AppProcessBackup) {
			defer wg.Done()
			errD := b.DeleteAppProcessBackup(&app_process)
			if errD != nil {
				log.RunLogger.Errorf("[express data cleaner] delete app process %s from backup table failed: %+v", app_process.ID, errD)
				return
			}
			log.RunLogger.Infof("[express data cleaner] delete app process %s from backup table success", app_process.ID)
		}(ap)
	}
	wg.Wait()
	return nil
}

// BackupAppProcess: 定期从运行表中备份残留的进程
func (b *AppProcessBackupDao) BackupAppProcess(dayBefore int) error {
	var aps []AppProcessBackup
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE UPDATED_AT < DATE_SUB(NOW(), INTERVAL %d DAY)",
		TableNameAppProcess, dayBefore)
	toBeBackupCount, err := b.sqlSession.Raw(sqlStr).QueryRows(&aps)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] query to be backup app process from runtime table failed: %+v", err)
		return err
	}
	log.RunLogger.Infof("[express data cleaner] find %v to be backup app process from runtime table", toBeBackupCount)
	wg := sync.WaitGroup{}
	wg.Add(int(toBeBackupCount))
	for idx, ap := range aps {
		if idx%AppProcessBackupCountToSleep == 0 {
			time.Sleep(AppProcessBackupCountToSleepSeconds * time.Second)
		}
		go func(app_process AppProcessBackup) {
			defer wg.Done()
			_, errI := b.InsertAppProcessBackup(&app_process)
			// 若插入失败，则可能是备份表中已存在该数据，尝试先从备份表中删除，再等待下一轮进行备份
			if errI != nil {
				log.RunLogger.Errorf("[express data cleaner] backup app process %s from runtime table to backup table failed: %+v, try to delete it from backup table",
					app_process.ID, errI)
				backupAp, err := b.QueryAppProcessBackupById(app_process.ID)
				if err != nil {
					log.RunLogger.Errorf("[express data cleaner] query app process %s from backup table failed: %+v",
						app_process.ID, errI)
					return
				}
				errD := b.DeleteAppProcessBackup(backupAp)
				if errD != nil {
					log.RunLogger.Errorf("[express data cleaner] delete app process %s from backup table failed: %+v",
						backupAp.ID, errI)
					return
				}
				log.RunLogger.Infof("[express data cleaner] delete app process %s from backup table success", backupAp.ID)
				return
			}
			sqlStrA := fmt.Sprintf("DELETE FROM %s WHERE ID = '%s'", TableNameAppProcess, app_process.ID)
			_, errD := b.sqlSession.Raw(sqlStrA).Exec()
			if errD != nil {
				log.RunLogger.Errorf("[express data cleaner] delete terminated app process %s from runtime table error: %+v", app_process.ID, errD)
				return
			}
			log.RunLogger.Infof("[express data cleaner] backup app process %s from runtime table to backup table success",
				app_process.ID)
		}(ap)
	}
	wg.Wait()
	return nil
}

// 从备份表中备份已被终止的进程
func (b *AppProcessBackupDao) BackupTerminatedAppProcess(Id string, wg *sync.WaitGroup) error {
	defer wg.Done()
	ap := &AppProcessBackup{}
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE ID='%s'", TableNameAppProcess, Id)
	err := b.sqlSession.Raw(sqlStr).QueryRow(ap)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] query terminated app process %s from runtime table error: %+v", Id, err)
		return err
	}
	// 将待清理数据插入备份表中
	_, err = b.InsertAppProcessBackup(ap)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] insert terminated app process %s into backup table error: %+v", Id, err)
		return err
	}
	// 将待清理数据从运行表中删除
	sqlStrA := fmt.Sprintf("DELETE FROM %s WHERE ID = '%s'", TableNameAppProcess, Id)
	_, err = b.sqlSession.Raw(sqlStrA).Exec()
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] delete terminated app process %s from runtime table error: %+v", Id, err)
		return err
	}
	log.RunLogger.Infof("[express data cleaner] success to backup terminated app process %s", Id)
	return nil
}
